package org.ultrapower.checkuser.core.util;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class MyPut
{
  static Log LOG = LogFactory.getLog(MyPut.class);

  public static void main(String[] args) throws Exception {
    long count = Long.parseLong(args[0]);
    long trigger = Integer.parseInt(args[1]);
    int clientNum = Integer.parseInt(args[2]);
    long start = System.currentTimeMillis();
    LOG.info("=====begin======");
    CountDownLatch runningThreadNum = new CountDownLatch(clientNum);
    for (int i = 0; i < clientNum; i++) {
      PutThread putThread = new PutThread(count, trigger, i, runningThreadNum);
      putThread.run();
    }
    try {
      runningThreadNum.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    long end = System.currentTimeMillis();
    float every = (float)((end - start) / (clientNum * count));
    LOG.info(clientNum + "clients put " + clientNum + "*" + count + " records corst " + (end - start) + " ms," + trigger + " records flush,every records corst " + every + " ms");
  }

  public static class MD5RowKeyGenerator
  {
    private final Logger LOG = Logger.getLogger(MD5RowKeyGenerator.class);

    private MessageDigest md = null;

    public String generate(String oriRowKey, String needHashValue, String[] currenRowdata, int[] posIndex, String appendValue)
    {
      return null;
    }

    public Object generate(String oriRowKey) {
      return generatePrefix(oriRowKey) + oriRowKey;
    }

    public synchronized String getMD5(String oriRowKey)
    {
      char[] hexDigits = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'b', 'd', 'e', 'f' };
      try {
        byte[] btInput = oriRowKey.getBytes();

        MessageDigest mdInst = MessageDigest.getInstance("MD5");

        mdInst.update(btInput);

        byte[] md = mdInst.digest();

        int j = md.length;
        char[] str = new char[j * 2];
        int k = 0;
        for (int i = 0; i < j; i++) {
          byte byte0 = md[i];
          str[(k++)] = hexDigits[(byte0 >>> 4 & 0xF)];
          str[(k++)] = hexDigits[(byte0 & 0xF)];
        }
        return new String(str);
      } catch (Exception e) {
      }
      return "";
    }

    public Object generatePrefix(String oriRowKey)
    {
      String result = getMD5(oriRowKey);
      return result.substring(1, 2) + result.substring(3, 4) + result.substring(5, 6);
    }
  }

  static class PutThread
    implements Runnable
  {
    static String tableName = "TESTREGIONREPLICATION";
    static String columnFamily = "F";
    static String[] cols = { "A", "B", "C", "D", "E", "F", "G", "H", "I", "J" };
    long count;
    long trigger;
    int client;
    CountDownLatch runningThreadNum;

    public PutThread(long count, long trigger, int client, CountDownLatch runningThreadNum)
    {
      this.count = count;
      this.trigger = trigger;
      this.client = client;
      this.runningThreadNum = runningThreadNum;
    }

    public void run()
    {
      Configuration conf = HBaseConfiguration.create();
      HTableInterface table = null;
      try {
        table = new HTable(conf, tableName);
        table.setAutoFlush(false, true);
        table.setWriteBufferSize(2097152L);
        List putList = new ArrayList();
        long start = System.currentTimeMillis();
        for (int i = 0; i <= this.count; i++) {
          Put put = new Put(Bytes.toBytes(new MD5RowKeyGenerator().getMD5(this.client + "_" + i)));
          for (int j = 0; j < cols.length; j++) {
            put.add(columnFamily.getBytes(), cols[j].getBytes(), (i + "" + j).getBytes());
          }
          putList.add(put);
          if (i % this.trigger == 0L) {
            table.put(putList);
            table.flushCommits();
          }
        }
        table.flushCommits();
      } catch (IOException e) {
        e.printStackTrace();
      } finally {
        this.runningThreadNum.countDown();
      }
    }
  }
}